/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.master;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.NTable;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.client.ScanCell;
import com.nhncorp.neptune.client.scanner.ScannerFactory;
import com.nhncorp.neptune.client.scanner.TableScanner;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.GlobalConstants;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.NStatusHttpServer;
import com.nhncorp.neptune.common.aop.ProxyFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.TableExistsException;
import com.nhncorp.neptune.common.ipc.AclManager;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.common.ipc.NRPC.Server;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.neptune.fs.ChangeLogFileSystem;
import com.nhncorp.neptune.fs.ChangeLogFileSystemIF;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.TableSchema;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.AsyncTask;
import com.nhncorp.neptune.tabletserver.AsyncTaskManager;
import com.nhncorp.neptune.tabletserver.AsyncTaskStatus;
import com.nhncorp.neptune.tabletserver.Tablet;
import com.nhncorp.neptune.tabletserver.TabletManagerProtocol;
import com.nhncorp.neptune.tabletserver.TabletServerInfo;
import com.nhncorp.pleiades.client.FilteredLockEventMonitor;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventFilter;
import com.nhncorp.pleiades.client.LockEventMonitor;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.protocol.LockMetaData;

/**
 * Neptune의 클러스터를 관리하는 Master<p>
 * NeptuneMaster의 역할은 Tablet을 할당해주는 역할만 수행한다. 
 * 따라서 NeptuneMaster에 장애가 발생해도 데이터 서비스(get, put)은 정상 서비스 가능하다.
 * 
 * @author 김형준
 * 
 */
public class NeptuneMaster implements NeptuneMasterIF {
  public static final Log LOG = LogFactory
      .getLog(NeptuneMaster.class.getName());

  private NConfiguration conf;

  private NeptuneFileSystem fs;

  /**
   * NeptuneMaster host 명(HOST명:PORT)
   */
  protected String hostName;

  /**
   * 요청을 받아 처리하는 메인 데몬 프로세스
   */
  private Server server;

  /**
   * 테이블 스키마 정보(tableName -> table)
   */
  private Map<String, TableSchema> tables;

  /**
   * 전체 Tablet 목록(tableName -> [tabletName -> tablets])
   */
  protected Map<String, Map<String, TabletInfo>> tabletInfos = new HashMap<String, Map<String, TabletInfo>>(
      100);

  /**
   * 할당되지 않은 Tablet 목록(TabletName -> TabletInfo)
   */
  private Map<String, TabletInfo> unassignedTablets = new HashMap<String, TabletInfo>(
      100);

  /**
   * 할당 진행 중인 Tablet 목록(TabletName -> TabletInfo)
   */
  private Map<String, TabletInfo> assigningTablets = new HashMap<String, TabletInfo>(
      100);

  /**
   * Live한 TabletServer 목록(hostName -> TabletServerInfo
   */
  private Map<String, TabletServerInfo> runningTabletServers = new HashMap<String, TabletServerInfo>(
      10);

  /**
   * 마지막으로 할당된 TabletServer(Tablet 할당 작업시 마지막에 할당된 TabletServer로는 할당 하지 않음)
   */
  private TabletServerInfo lastAssignedTabletServer;

  /**
   * Drop 진행중인 TabletServer 목록(TableName -> TabletServer HostName)
   */
  private Map<String, List<String>> dropingTabletServers = new HashMap<String, List<String>>(
      10);

  /**
   * Tablet 할당을 위해 TabletServer를 선택하기 위한 Random 객체
   */
  // private Random random = new Random();
  /**
   * Master startup시 Root Tablet이 할당 완료 여부
   */
  private boolean endRootTableAssignment = false;

  /**
   * Tablet 관련 변수 추가, 삭제시 lock monitor
   */
  private Object tabletInfoMonitor = new Object();

  /**
   * NeptuneMaster 서버 모든 Thread 종료 flag
   */
  private boolean stopRequested = false;

  /**
   * Root, Meta loading이 완료될 경우 true
   */
  private static boolean clusterReady = false;

  /**
   * AOP를 이용한 테스트 코드 작성을 위해 필요한 메소드 호출을 Proxy를 통하도록 한다.
   */
  private NeptuneMasterIF proxyObject;

  /**
   * Pleiades lock service
   */
  private LockService lockService;

  private ThreadGroup threadGroup;

  private NStatusHttpServer infoServer;

  private int infoPort;

  public static NeptuneMaster neptuneMaster;

  protected Date masterStartTime;

  protected Date masterInitTime;

  //protected String version = "0.0.1-dev";

  protected AsyncTaskManager asyncTaskManager = new AsyncTaskManager();

  protected int heartbeatPort;
  
  public NeptuneMaster() {
    System.setProperty("java.net.preferIPv4Stack", "true");
    neptuneMaster = this;
    masterStartTime = new Date();
  }

  public void init(NeptuneMasterIF proxyObject, NConfiguration conf)
      throws IOException {
    // try {
    // Thread.sleep(10 * 1000);
    // } catch (InterruptedException e) {
    // return;
    // }
    InetSocketAddress serverAddress = NetworkUtil.getLocalAddress(conf.getInt(
        "masterServer.port", 7000));
    this.hostName = serverAddress.getHostName() + ":" + serverAddress.getPort();

    LOG.info("NeptuneMaster initialization at " + hostName);

    this.threadGroup = new ThreadGroup("NeptuneMaster_" + hostName);

    this.conf = conf;

    this.proxyObject = proxyObject;
    this.fs = NeptuneFileSystem.get(conf);

    this.heartbeatPort = conf.getInt("masterServer.heartbeat.port", 50200);
    
    this.lockService = LockServiceFactory.getLockService(conf, hostName, true, heartbeatPort);
    
    this.server = NRPC.getServer(lockService, proxyObject, serverAddress
        .getHostName(), serverAddress.getPort(), conf.getInt(
        "masterServer.handler.count", 10), false, conf);

    this.server.start();
  }

  /**
   * NeptuneMaster main thread
   */
  public void run() {
    try {
      while (true) {
        try {
          boolean lock = acquireMasterLock();
          if (lock) {
            break;
          }
          Thread.sleep(3000);
        } catch (IOException ie) {
          LOG.error(ie.getMessage(), ie);
        } catch (InterruptedException e) {
          return;
        }
      }

      // Master 초기화 작업
      LOG.info("NeptuneMaster[" + hostName + "] get master lock.");
      masterInit();
    } catch (Exception e) {
      LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
      shutdown();
    }
  }

  /**
   * NeptuneMaster의 기능을 수행하기 위해 Pleiades로 부터 lock을 얻는다.
   * 
   * @return
   * @throws IOException
   */
  private boolean acquireMasterLock() throws IOException {
    LockMetaData lockMetaData = new LockMetaData(Constants.MASTER_SERVER);
    lockMetaData.setContentKey(hostName);
    lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
    if (lockService == null) {
      LOG.debug("LockService closed: " + hostName);
      return false;
    }
    boolean locked = lockService.createNode(lockMetaData, true);
    // LOG.debug("-------------------- Neptune Master is Waiting from : " +
    // getHostName() + " --------------------" + locked);
    return locked;
  }

  public boolean tryMasterLock() throws IOException {
    return lockService.tryLock(Constants.MASTER_SERVER,
        LockService.LockMode.LOCK_EXCLUSIVE);
  }

  /**
   * Master가 lock을 acquire한 다음의 작업을 수행한다.
   * 
   * @throws IOException
   */
  private void masterInit() throws IOException {
    masterInitTime = new Date();
    setClusterReady(false);
//    try {
//      Thread.sleep(2 * 1000);
//    } catch (InterruptedException e1) {
//      return;
//    }

    try {
      tables = loadAllTableSchemas(conf, lockService);
    } catch (IOException e) {
      LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
      shutdown();
      return;
    }

    addLockEventHandler();

    Thread rootTabletAssignmentThread = new Thread(threadGroup,
        new RootTabletAssignmentThread());
    rootTabletAssignmentThread.start();

    infoPort = conf.getInt("master.info.port", 57000);
    InetSocketAddress infoServerAddress = NetworkUtil.getAddress(InetAddress
        .getLocalHost().getHostName()
        + ":" + infoPort);

    String infoHost = infoServerAddress.getHostName();
    try {
      this.infoServer = new NStatusHttpServer("master", infoHost, infoPort);
      this.infoServer.start();
    } catch (Exception e) {
      LOG.warn("Error while info server init:" + e.getMessage());
    }

    Thread lockChecker = new Thread(threadGroup, new MasterServerLockChecker());
    lockChecker.start();

    // 종료되지 않은 tablet drop에 대한 처리
    Map<String, LockMetaData> dropLocks = lockService
        .readDir(Constants.TABLE_DROP);
    if (dropLocks != null) {
      for (LockMetaData lockMetaData : dropLocks.values()) {
        String dropTableName = LockServiceFactory
            .getLockMetaDataPathName(lockMetaData);
        asyncTaskManager.runAsyncTask(new TableDropTask(dropTableName));
      }
    }
  }

  /**
   * TabletServer로 접속한다.
   * 
   * @param hostName
   * @return
   * @throws IOException
   */
  protected TabletManagerProtocol connectTabletServer(String hostName)
      throws IOException {
    return connectTabletServer(conf, hostName);
  }

  static TabletManagerProtocol connectTabletServer(NConfiguration conf,
      String hostName) throws IOException {
    // LOG.debug("connectTabletServer:hostname=" + hostName);
    TabletManagerProtocol tabletServer = (TabletManagerProtocol) NRPC.getProxy(
        TabletManagerProtocol.class, TabletManagerProtocol.versionID,
        NetworkUtil.getAddress(hostName), conf);

    return tabletServer;
  }

  private TabletManagerProtocol connectTabletServer(
      TabletServerInfo tabletServerInfo) throws IOException {
    return connectTabletServer(tabletServerInfo.getHostName());
  }

  /**
   * ROOT, META 테이블의 생성/로딩 및 각각의 첫 Tablet이 생성/로딩이 완료되는 시점에 ready가 true로 변경된다.
   * 
   * @param ready
   */
  private void setClusterReady(boolean ready) throws IOException {
    clusterReady = ready;
  }

  /**
   * ROOT, META가 모두 로딩된 상태
   * 
   * @return
   * @throws IOException
   */
  public static boolean isClusterReady() throws IOException {
    return clusterReady;
  }

  /**
   * META tablet을 할당한다. META table이 없으면 Table을 생성한다.
   * 
   * @throws IOException
   */
  private void assignMetaTablet(Map<String, TabletInfo> assignedTablets)
      throws IOException {
    // if meta not exists, make meta table
    boolean createdMetaTable = false;
    if (!tables.containsKey(TABLE_NAME_META)) {
      createTable(GlobalConstants.META_TABLE, null);
      createdMetaTable = true;
    }

    Map<String, TabletInfo> metaTablets = tabletInfos.get(TABLE_NAME_META);

    LOG.debug("createdMetaTable : " + createdMetaTable
        + ", metaTablets.size() : " + metaTablets.size());
    if (createdMetaTable && metaTablets.size() == 0) {
      // if empty, create
      // MaxTableName.MaxRow.Key
      TabletInfo tabletInfo = new TabletInfo(TABLE_NAME_META, Tablet
          .generateTabletName(TABLE_NAME_META), Row.Key.MIN_KEY, Tablet
          .generateMetaRowKey(null, Row.Key.MAX_KEY));

      LOG.debug("Create META tablet:" + tabletInfo);
      synchronized (tabletInfoMonitor) {
        metaTablets.put(tabletInfo.getTabletName(), tabletInfo);
        unassignedTablets.put(tabletInfo.getTabletName(), tabletInfo);
        proxyObject.assignTablet(tabletInfo);
      }
    } else {
      LOG.debug("Not create META tablet");
      List<TabletInfo> tempMetaTablets = new ArrayList<TabletInfo>();
      synchronized (tabletInfoMonitor) {
        tempMetaTablets.addAll(metaTablets.values());
      }
      for (TabletInfo tabletInfo : tempMetaTablets) {
        String tabletName = tabletInfo.getTabletName();

        // 현재 할당 중인 Tablet에 없으면 할당 처리
        if (!assignedTablets.containsKey(tabletName)) {
          synchronized (tabletInfoMonitor) {
            unassignedTablets.put(tabletInfo.getTabletName(), tabletInfo);
            try {
              proxyObject.assignTablet(tabletInfo);
            } catch (IOException e) {
              LOG.warn(e);
              // assignTablet 작업시 오류에 대한 처리 수행
            }
          }
        }
      }
    }
  }

  /**
   * ROOT tablet을 할당한다. ROOT table이 없으면 Table을 생성한다.
   * 
   * @return
   * @throws IOException
   */
  private boolean assignRootTablet() throws IOException {
    // if root not exists, make root table
    TabletInfo rootTabletInfo = null;
    if (!tables.containsKey(TABLE_NAME_ROOT)) {
      // LOG.info("Create ROOT Tablet");
      // ROOT user 존재확인 후 존재하지 않으면 생성
      if (!existSuperGroup()) {
        LOG.info("Create supergroup user(" + conf.getUserId() + ")");
        addUser(lockService, conf.getUserId());
        addSuperGroupUser(lockService, conf.getUserId());
      }

      // 존재하지 않으면 생성
      createTable(GlobalConstants.ROOT_TABLE, null);
      rootTabletInfo = new TabletInfo(TABLE_NAME_ROOT, Tablet
          .generateTabletName(TABLE_NAME_ROOT), Row.Key.MIN_KEY,
          Row.Key.MAX_KEY);
      LockMetaData lockMetaData = new LockMetaData(Constants.ROOT_TABLET);
      lockMetaData.setContentKey(rootTabletInfo.getTabletName());
      lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
      boolean locked = lockService.createNode(lockMetaData, true);
      if (!locked) {
        LOG.fatal("NeptuneMaster shutdown cause:Fail ROOT tablet lock:"
            + Constants.ROOT_TABLET);
        shutdown();
        return false;
      }

//      GPath rootTabletInfoPath = new GPath(conf.get("neptune.root") + Constants.ROOT_TABLET + "/" + rootTabletInfo.getTabletName());
//      boolean success = fs.mkdirs(rootTabletInfoPath);
//      if(!success) {
//        LOG.fatal("NeptuneMaster shutdown cause:Fail create root tablet info in dfs:" 
//            + rootTabletInfoPath);
//        shutdown();
//        return false;
//      }
//      boolean locked = LockServiceFactory.getLockService(conf, hostName, true).getLock(
//          Constants.ROOT_TABLET,
//          LockService.LockMode.LOCK_EXCLUSIVE);
//      
//      if (!locked) {
//        LOG.fatal("NeptuneMaster shutdown cause:Fail ROOT tablet lock while table creating:"
//            + Constants.ROOT_TABLET);
//        shutdown();
//        return false;
//      }
      
      lockService.releaseLock(Constants.ROOT_TABLET,
          LockService.LockMode.LOCK_EXCLUSIVE);
    } else {
      // 존재하면 TabletInfo 로딩
      rootTabletInfo = Tablet.getRootTabletInfo(conf, lockService);
    }
    if (rootTabletInfo == null) {
      LOG.fatal("NeptuneMaster shutdown cause:Can't get ROOT tablet info");
      shutdown();
      return false;
    }

    // TabletServer가 존재하지 않으면 계속 반복
    synchronized (runningTabletServers) {
      if (runningTabletServers.size() == 0) {
        LOG.warn("No live tablet server while root tablet assign");
        return false;
      }
    }

    synchronized (tabletInfoMonitor) {
      Map<String, TabletInfo> rootTablets = new HashMap<String, TabletInfo>();
      rootTablets.put(rootTabletInfo.getTabletName(), rootTabletInfo);
      tabletInfos.put(TABLE_NAME_ROOT, rootTablets);
    }

    // TabletServer는 stop 되지 않고 Master만 kill또는 stop 된 경우
    // 이전에 서비스 중인 RootTablet에 대한 lock을 가져온다.
    // lock을 가져올 수 있으면 다시 할당하고 lock을 못가져 오면 이미 TabletServer에서 서비스 중이기 때문에
    // 아무런 조치를 하지 않는다.
    boolean locked = lockService.getLock(Constants.ROOT_TABLET,
        LockService.LockMode.LOCK_EXCLUSIVE);
    // LOG.debug("ROOT Tablet lock:" + locked);
    if (locked && !assigningTablets.containsKey(rootTabletInfo.getTabletName())) {
      // lock이 성공하고 이미 할당 중인 경우가 아니라면 Root를 다른 TabletServer에 할당
      lockService.releaseLock(Constants.ROOT_TABLET,
          LockService.LockMode.LOCK_EXCLUSIVE);
      unassignedTablets.put(rootTabletInfo.getTabletName(), rootTabletInfo);
      return proxyObject.assignTablet(rootTabletInfo);
    } else if(!locked){
      LOG.warn("Can't get ROOT Tablet Lock:" + Constants.ROOT_TABLET);
      return false;
    } else if(assigningTablets.containsKey(rootTabletInfo.getTabletName())){
      LOG.info("ROOT tablet already exists in assigningTablets");
      endRootTableAssignment = true;
      return true;
    } else {
      LOG.warn("No condition");
      return false;
    }
    
  }

  /**
   * 테이블 스키마를 로딩한다. 테이블 스키마는 DFS상의 특정 디렉토리 아래에 하위 디렉토리로 명으로 저장된다. 테이블명/컬러명1
   * /컬럼명2
   * 
   * @param conf
   * @return
   * @throws IOException
   */
  private static Map<String, TableSchema> loadAllTableSchemas(NConfiguration conf, LockService lockService) throws IOException {
    String schemaLocation = conf.get("table.schema.type", "filesystem");
    if(schemaLocation.equals("pleiades")) {
      return loadAllTableSchemasFromLockService(conf, lockService);
    } else {
      return loadAllTableSchemasFromFile(conf, lockService);
    }
  }
  
  private static Map<String, TableSchema> loadAllTableSchemasFromFile(NConfiguration conf, LockService lockService) throws IOException {
    Map<String, TableSchema> tables = new HashMap<String, TableSchema>(10);
    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
    GPath schemaPath = new GPath(conf.get("neptune.root") + PATH_SCHEMA);
    GPath[] tablePaths = fs.list(schemaPath);

    if (tablePaths == null || tablePaths.length == 0)
      return tables;

    for (GPath eachPath : tablePaths) {
      String tableName = eachPath.getName();
      TableSchema table = TableSchema.loadTableSchema(conf, lockService, tableName);
      // LOG.debug("Load Table schema:" + table.getTableName());
      tables.put(tableName, table);
    }

    return tables;
  }
  
  public static Map<String, TableSchema> loadAllTableSchemasFromLockService(NConfiguration conf, LockService lockService) throws IOException {
    Map<String, TableSchema> tables = new HashMap<String, TableSchema>(10);
    Map<String, LockMetaData> metadatas = lockService.readDir(Constants.PATH_SCHEMA,
        false);

    if (metadatas == null) {
      return tables;
    }
    for (LockMetaData eachLockMetaData : metadatas.values()) {
      String tableName = LockServiceFactory
          .getLockMetaDataPathName(eachLockMetaData);
      TableSchema table = TableSchema.loadTableSchema(conf, lockService, tableName);
      tables.put(table.getTableName(), table);
    }

    return tables;
  }

  /**
   * master가 실행되고 있는 서버 정보를 반환한다.
   * 
   * @param conf
   * @return
   */
  public static String getMasterServerHostName(LockService lockService)
      throws IOException {
    LockMetaData lockMetaData = lockService
        .getMetaData(Constants.MASTER_SERVER);
    if (lockMetaData == null) {
      return null;
    }
    String masterHostName = lockMetaData.getContentKey();
    return masterHostName;
  }

  /**
   * 모든 데이터 및 스키마를 삭제한다.
   * 
   * @throws IOException
   */
  public static void format() throws IOException {
    System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
    System.out.println("Warning!!!!");
    System.out.println("Format will remove all datas!");
    System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
    System.out.print("Continue format(Y|N): ");
    BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
    String answer = reader.readLine();
    if ("Y".equals(answer)) {
      System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
      System.out.println("Start format");
      NConfiguration conf = new NConfiguration();
      doFormat(conf);
      System.out.println("End format");
      System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
    } else {
      System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
      System.out.println("format is cancelled");
      System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
    }
    return;
  }

  public static void formatFileSystem(NConfiguration conf)
      throws IOException {
    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
    
    if (!fs.delete(new GPath(conf.get("neptune.root")), true)) {
      LOG.warn("Can't all data file:" + conf.get("neptune.root"));
    }
    // fs.close();
  }

  /**
   * 모든 데이터 및 스키마를 삭제한다.
   * 
   * @throws IOException
   */
  public static boolean doFormat(NConfiguration conf) throws IOException {
    LOG.debug("start deleting files and lock");
    formatFileSystem(conf);
    LOG.debug("end deleting files");

    long startTime = System.currentTimeMillis();

    // 이전의 모든 lock 정보를 삭제할때 까지 반복 수행
    while (true) {
      try {
        Thread.sleep(5 * 1000);
      } catch (InterruptedException e1) {
        return true;
      }

      LockService lockService = null;
      try {
        lockService = LockServiceFactory.getLockService(conf, "NeptuneMaster", false);

        if (lockService != null) {
          ChangeLogFileSystemIF changeLogFileSystem = ChangeLogFileSystem
              .getChangeLogFileSystem(conf, null, lockService);
          LOG.debug("deleting changelog");
          changeLogFileSystem.format();

          if (!lockService.exist(Constants.NEPTUNE)) {
            // 아무것도 없는 경우
            LOG.debug("no lock node");
            return true;
          } else {
            boolean lockResult = lockService
                .deleteNode(Constants.NEPTUNE, true);
            if (lockResult) {
              LOG.debug("deleted files and lock");
              lockService.releaseAll();
              return true;
            }
          }
        } else {
          LockServiceFactory.clear("NeptuneMaster");
        }

        if (System.currentTimeMillis() - startTime > 30 * 1000) {
          LOG.error("Can't format. timeout");
          return false;
        }
      } catch (Exception e) {
        LOG.error(e.getMessage(), e);
      } finally {
        if (lockService != null) {
          NeptuneMaster.addUser(lockService, conf.getUserId());
          NeptuneMaster.addSuperGroupUser(lockService, conf.getUserId());
          LockServiceFactory.close("NeptuneMaster");
        } else {
          LOG.fatal("Can't get LockServer: lock opwner=NeptuneMaster");
        }
      }
    }
  }

  /**
   * 
   * @param args
   * @throws IOException
   */
  public static void main(String args[]) throws IOException {
    if (args.length > 0) {
      if ("-format".equals(args[0])) {
        format();
      } else if ("-drop".equals(args[0])) {
        String tableName = args[1];
        forcedDropTable(tableName);
      } else {
        System.out.println("use '-format'");
      }
      System.exit(0);
    }
    try {
      NeptuneMasterIF masterServer = (NeptuneMasterIF) ProxyFactory
          .createProxyClass("masterServer");
      NConfiguration conf = new NConfiguration();

      masterServer.init(masterServer, conf);
      (new Thread(masterServer)).start();
    } catch (Exception e) {
      LOG.fatal("Error while start NeptuneMaster:" + e.getMessage(), e);
    }
  }

  /**
   * lock, schema 존재여부 상관없이 drop 처리
   * 
   * @param tableName
   * @throws IOException
   */
  public static void forcedDropTable(String tableName) throws IOException {
    NConfiguration conf = new NConfiguration();
    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);

    LockService lockService = LockServiceFactory.getLockService(conf,
        "NeptuneMaster", false);

    String lockPath = Constants.TABLE_DROP + "/" + tableName;
    LockMetaData lockMetaData = new LockMetaData(lockPath);
    lockMetaData.setContentKey(String.valueOf(0));
    lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());

    boolean locked = lockService.createNode(lockMetaData, true);

    finalizeDrop(null, conf, fs, lockService, tableName, true);
  }

  /**
   * 테이블에 컬럼을 추가한다.
   */
  public void addColumn(String tableName, String addedColumnName)
      throws IOException {
    // FIXME lock 처리 추가
    synchronized (tables) {
      AclManager.checkPermission(conf, lockService, tables, tableName, "r");

      if (!tables.containsKey(tableName)) {
        throw new IOException("Table not exists [" + tableName + "]");
      }

      TableSchema tableSchema = tables.get(tableName);
      if (tableSchema.getColumns().contains(addedColumnName)) {
        throw new IOException("Column already exists [" + addedColumnName + "]");
      }
      tableSchema.addColumn(addedColumnName);

      // Pleiades lock 파일 생성
//      String tablePath = Constants.SCHEMA + "/" + tableName;
//      LockMetaData columnLockMetaData = new LockMetaData(tablePath + "/"
//          + addedColumnName);
//      columnLockMetaData.setContentKey(String.valueOf(tableSchema.getColumns()
//          .size()));
//      lockService.createNode(columnLockMetaData, true);
      GPath columnPath = new GPath(TableSchema.getTableSchemaPath(conf, tableName), addedColumnName);
      fs.mkdirs(columnPath);
    }
  }

  public TabletInfo addTablet(String tableName, Row.Key endRowKey)
      throws IOException {
    if (tableName == null || tableName.equals(TABLE_NAME_ROOT)
        || tableName.equals(TABLE_NAME_META)) {
      throw new IOException("Check table name:" + tableName);
    }

    if (endRowKey == null || endRowKey.equals(Row.Key.MAX_KEY)
        || endRowKey.equals(Row.Key.MIN_KEY)) {
      throw new IOException("Check endRowKey:" + endRowKey);
    }

    synchronized (tabletInfoMonitor) {
      if (!tables.containsKey(tableName)) {
        throw new IOException("No Table:" + tableName);
      }
      Row.Key metaRowKey = Tablet.generateMetaRowKey(tableName, endRowKey);
      NTable ntable = NTable.openTable(conf, TABLE_NAME_META);
      if (ntable.hasValue(META_COLUMN_NAME_TABLETINFO, metaRowKey)) {
        throw new IOException("Already exists tablet:" + tableName + ","
            + endRowKey);
      }

      String tabletName = Tablet.generateTabletName(tableName);
      TabletInfo tabletInfo = new TabletInfo(tableName, tabletName);
      tabletInfo.setEndRowKey(endRowKey);
      proxyObject.assignTablet(tabletInfo);

      return tabletInfo;
    }
  }

  /**
   * 테이블을 생성한다. endRowKeys에 초기에 생성할 tablet에 대한 목록이 있는 경우 tablet을 여러개 생성시키고 그렇지
   * 않는 경우 min ~ max 범위를 저장하는 하나의 tablet만 생성한다.
   * 
   * @param table
   * @param endRowKeys
   */
  public void createTable(TableSchema table, Row.Key[] endRowKeys) throws IOException {
    synchronized (tables) {
      if (endRowKeys != null && endRowKeys.length > 0
            && !endRowKeys[endRowKeys.length - 1].equals(Row.Key.MAX_KEY)) {
        throw new IOException("End rowkey must max value");
      }
      
      // make table schema
      String tableName = table.getTableName();
      LOG.debug("Create Table:" + tableName);
      
      String tabletDropLockPath = Constants.TABLET_DROP + "/" + tableName;
      if (lockService.getMetaData(tabletDropLockPath) != null) {
        throw new TableExistsException("table [" + tableName + "] dropping");
      }    
      
      if (tables.containsKey(tableName)) {
        throw new TableExistsException("already exists table [" + tableName + "]");
      }
      
      saveTableSchema(table);
      
      tables.put(tableName, table);
      
      synchronized (tabletInfoMonitor) {
        tabletInfos.put(tableName, new HashMap<String, TabletInfo>(100));
      }

      // assign default tablet
      // default tablet row range : 0 ~ 무한대
      if (!TABLE_NAME_ROOT.equals(tableName)
          && !TABLE_NAME_META.equals(tableName)) {
        if (!tabletInfos.containsKey(tableName)) {
          tabletInfos.put(tableName, new HashMap<String, TabletInfo>(100));
        }
        if (endRowKeys == null || endRowKeys.length == 0) {
          endRowKeys = new Row.Key[1];
          endRowKeys[0] = Row.Key.MAX_KEY;
        }
        Row.Key startRowKey = Row.Key.MIN_KEY;
        for (int i = 0; i < endRowKeys.length; i++) {
          String tabletName = Tablet.generateTabletName(tableName);
          TabletInfo tabletInfo = new TabletInfo(tableName, tabletName,
              startRowKey, endRowKeys[i]);
          proxyObject.assignTablet(tabletInfo);
          startRowKey = endRowKeys[i];
        }
      }
    }
  }

  private void saveTableSchema(TableSchema table)
      throws IOException, TableExistsException {
    String schemaLocation = conf.get("table.schema.type", "filesystem");
    if(schemaLocation.equals("pleiades")) {
      saveTableSchemaToLockService(table);
    } else {
      saveTableSchemaToFile(table);
    }    
  }
  
  private void saveTableSchemaToLockService(TableSchema table) throws IOException, TableExistsException {
    String tableName = table.getTableName();
    
    String tablePath = Constants.PATH_SCHEMA + "/" + tableName;
    // Table 정보 저장
    LockMetaData detailLockMetaData = new LockMetaData(tablePath
        + SCHEMA_DETAIL);
    detailLockMetaData.setContentKey(String.valueOf(table.getNumOfVersion()));
    lockService.createNode(detailLockMetaData, true);
    lockService.setContents(tablePath + SCHEMA_DETAIL, table.getOwnerInfo());
  
    // Column 저장
    int columnIndex = 1;
    for (String column : table.getColumns()) {
      LockMetaData columnLockMetaData = new LockMetaData(tablePath + "/"
          + column);
      columnLockMetaData.setContentKey(String.valueOf(columnIndex++));
      lockService.createNode(columnLockMetaData, true);
    }
  
    String descPath = tablePath + "/"
        + Constants.SCHEMA_DESCRIPTION_FILE_NAME;
    lockService.setContents(descPath, table.getDescription());
  }
  
  private void saveTableSchemaToFile(TableSchema table) throws IOException, TableExistsException {
    String tableName = table.getTableName();
    
    GPath tableSchemaPath = TableSchema.getTableSchemaPath(conf, tableName);
    if (fs.exists(tableSchemaPath)) {
      throw new TableExistsException("already exists table [" + tableName + "]");
    }

    fs.mkdirs(tableSchemaPath);

    GPath schemaInfoPath = new GPath(tableSchemaPath, SCHEMA_INFO_FILENAME);
    OutputStream out = null;
    try {
      out = fs.create(schemaInfoPath);
      out.write(String.valueOf(table.getNumOfVersion()).getBytes());
      out.write("\n".getBytes());
      out.write(table.getOwner().getBytes());
      out.flush();
    } finally {
      if (out != null)
        out.close();
    }
    
    GPath descriptionPath = new GPath(tableSchemaPath, SCHEMA_DESCRIPTION_FILE_NAME);
    OutputStream dout = null;
    try {
      dout = fs.create(descriptionPath);
      if (table.getDescription() != null) {
        dout.write(table.getDescription().getBytes());
        dout.flush();
      }
    } finally {
      if (dout != null)
        dout.close();
    }

    for (String column : table.getColumns()) {
      GPath columnPath = new GPath(tableSchemaPath, column);
      fs.mkdirs(columnPath);
    }
  }
  
//  public void createTable(TableSchema table, Row.Key[] endRowKeys)
//      throws IOException {
//    synchronized (tables) {
//      if (endRowKeys != null && endRowKeys.length > 0
//          && !endRowKeys[endRowKeys.length - 1].equals(Row.Key.MAX_KEY)) {
//        throw new IOException("End rowkey must max value");
//      }
//      String tableName = table.getTableName();
//      // LOG.info("Create Table:" + tableName);
//
//      // 현재 drop 처리 중인 테이블인지 확인
//      String tabletDropLockPath = Constants.TABLET_DROP + "/" + tableName;
//      if (lockService.getMetaData(tabletDropLockPath) != null) {
//        throw new TableExistsException("table [" + tableName + "] dropping");
//      }
//
//      // make table schema
//      if (tables.containsKey(tableName)) {
//        throw new TableExistsException("already exists table [" + tableName
//            + "]");
//      }
//
//      String tablePath = Constants.SCHEMA + "/" + tableName;
//
//      // Table 정보 저장
//      LockMetaData detailLockMetaData = new LockMetaData(tablePath
//          + SCHEMA_DETAIL);
//      detailLockMetaData.setContentKey(String.valueOf(table.getNumOfVersion()));
//      lockService.createNode(detailLockMetaData, true);
//      lockService.setContents(tablePath + SCHEMA_DETAIL, table.getOwnerInfo());
//
//      // Column 저장
//      int columnIndex = 1;
//      for (String column : table.getColumns()) {
//        LockMetaData columnLockMetaData = new LockMetaData(tablePath + "/"
//            + column);
//        columnLockMetaData.setContentKey(String.valueOf(columnIndex++));
//        lockService.createNode(columnLockMetaData, true);
//      }
//
//      String descPath = tablePath + "/"
//          + Constants.SCHEMA_DESCRIPTION_FILE_NAME;
//      lockService.setContents(descPath, table.getDescription());
//
//      tables.put(tableName, table);
//
//      synchronized (tabletInfoMonitor) {
//        tabletInfos.put(tableName, new HashMap<String, TabletInfo>(100));
//      }
//
//      // assign default tablet
//      // default tablet row range : 0 ~ 무한대
//      if (!TABLE_NAME_ROOT.equals(tableName)
//          && !TABLE_NAME_META.equals(tableName)) {
//        if (!tabletInfos.containsKey(tableName)) {
//          tabletInfos.put(tableName, new HashMap<String, TabletInfo>(100));
//        }
//        if (endRowKeys == null || endRowKeys.length == 0) {
//          endRowKeys = new Row.Key[1];
//          endRowKeys[0] = Row.Key.MAX_KEY;
//        }
//        Row.Key startRowKey = Row.Key.MIN_KEY;
//        for (int i = 0; i < endRowKeys.length; i++) {
//          String tabletName = Tablet.generateTabletName(tableName);
//          TabletInfo tabletInfo = new TabletInfo(tableName, tabletName,
//              startRowKey, endRowKeys[i]);
//          proxyObject.assignTablet(tabletInfo);
//          startRowKey = endRowKeys[i];
//        }
//      }
//    }
//  }

  /**
   * Tablet을 tablet server에 할당한다. ignoreFlag는 tablet이 현재 다른 tabletserver에 의해
   * 로딩중이거나 서비스중에 상관없이 지정된 tablet server로 할당된다.
   */
  public boolean assignTablet(TabletInfo tabletInfo) throws IOException {
     LOG.info("assign tablet:" + tabletInfo);
    synchronized (tabletInfoMonitor) {
      if (assigningTablets.containsKey(tabletInfo)) {
        LOG.info("Already requested assigning(" + tabletInfo + ")");
        return true;
      }
      unassignedTablets.remove(tabletInfo.getTabletName());
      assigningTablets.put(tabletInfo.getTabletName(), tabletInfo);
    }

    int retry = 0;
    // 할당 가능한 TabletServer가 있을때까지 반복 수행
    while (retry < 5) {
      int runningSize = 0;
      synchronized (runningTabletServers) {
        runningSize = runningTabletServers.size();
      }
      if (runningSize == 0) {
        LOG
            .debug("no active tablet server. can't assign tablet: "
                + tabletInfo);
        break;
      }

      TabletServerInfo tabletServerInfo = null;
      try {
        long startTime = System.currentTimeMillis();
        while (true) {
          tabletServerInfo = selectTabletServerForAssignment();

          if (tabletServerInfo != null) {
            break;
          }
          if (System.currentTimeMillis() - startTime > (10 * 1000)) {
            break;
          }
        }
        if (tabletServerInfo == null) {
          LOG.error("Can't find proper tablet server(live TabletServer="
              + runningSize);
          retry++;
          continue;
        }
        TabletManagerProtocol tabletServer = connectTabletServer(tabletServerInfo);

        try {
          tabletServer.assignTablet(tabletInfo);
        } catch (IOException e) {
          //장애가 발생한 경우 해당 TabletServer에서 할당되었는지 다시 확인
          //FIXME 하나의 Tablet이 하나의 TabletServer에만 할당될 수 있도록 하는 정책 필요
          int errorRetry = 0;
          while(true)
          try {
            TabletInfo assignedTabletInfo = tabletServer.getTabletInfo(tabletInfo.getTabletName());
            if(assignedTabletInfo != null) {
              //정상적으로 할당되었음
              break;
            }
          } catch (IOException err) {
            LOG.warn("Error checking Tablet already assigned:" + tabletInfo + "," + tabletServerInfo);
            Thread.sleep(1000);
            errorRetry++;
            if(errorRetry >= 10) {
              throw new IOException(e);
            }
          }
        }

        tabletInfo.setAssignedHostName(tabletServerInfo.getHostName());

        tabletServerInfo.addNumOfTablets();

        LOG.debug("assignTablet: tabletName=" + tabletInfo.getTabletName()
            + ", assignedHost=" + tabletServerInfo.getHostName());
        return true;
      } catch (Exception e) {
        LOG.warn("error while assignment. but retry:" + e.getMessage());
        if (tabletServerInfo != null) {
          LOG.warn("Exception in assigning tablet : "
              + tabletInfo.getTabletName() + " to host : "
              + tabletServerInfo.getHostName() + ". Retry count : "
              + (retry + 1), e);
        } else {
          LOG.warn("Exception in assigning tablet : "
              + tabletInfo.getTabletName() + " to host null. Retry count : "
              + (retry + 1), e);
        }
        retry++;
        try {
          Thread.sleep(1 * 1000);
        } catch (InterruptedException e1) {
        }
      }
    }

    // if assignment fail
    synchronized (tabletInfoMonitor) {
      assigningTablets.remove(tabletInfo.getTabletName());
      unassignedTablets.put(tabletInfo.getTabletName(), tabletInfo);
    }
    LOG.debug("Assignment fail:" + tabletInfo);
    return false;
  }

  private TabletServerInfo selectTabletServerForAssignment() {
    List<TabletServerInfo> sortedTabletServers = new ArrayList<TabletServerInfo>();
    synchronized (runningTabletServers) {
      sortedTabletServers.addAll(runningTabletServers.values());
    }
    Collections.sort(sortedTabletServers,
        new TabletServerInfo.TabletServerInfoComparator());

    TabletServerInfo selectedServer = sortedTabletServers.get(0);

    if (lastAssignedTabletServer == null) {
      lastAssignedTabletServer = selectedServer;
      return selectedServer;
    } else {
      if (sortedTabletServers.size() > 1
          && selectedServer.equals(lastAssignedTabletServer)) {
        selectedServer = sortedTabletServers.get(1);
        lastAssignedTabletServer = selectedServer;
        return selectedServer;
      } else {
        lastAssignedTabletServer = selectedServer;
        return selectedServer;
      }
    }
  }

  public AsyncTaskStatus getAsyncTaskStatus(String taskId) throws IOException {
    return asyncTaskManager.getAsyncTaskStatus(taskId);
  }

  public void removeAsyncTask(String taskId) throws IOException {
    asyncTaskManager.removeAsyncTask(taskId);
  }

  public String dropTable(String tableName) throws IOException {
    AclManager.checkOwner(conf, lockService, tables, tableName);
    String taskId = asyncTaskManager.runAsyncTask(new TableDropTask(tableName));
    return taskId;
  }

  class TableDropTask extends AsyncTask {
    String tableName;

    public TableDropTask(String tableName) {
      this.tableName = tableName;
    }

    // FIXME Drop 도중에 Pleiades에서 schema 정보는 삭제 했는데 META에 삭제되지 않은 경우 처리
    public void exec() throws Exception {
      LOG.debug("Drop table:" + tableName);

      synchronized (dropingTabletServers) {
        if (dropingTabletServers.containsKey(tableName)
            && dropingTabletServers.get(tableName).size() > 0) {
          throw new IOException("Already dropping:" + tableName);
        }
      }

      // pleiades lock 생성
      String lockPath = Constants.TABLE_DROP + "/" + tableName;

      LockMetaData lockMetaData = lockService.getMetaData(lockPath);

      if (lockMetaData != null) {
        boolean locked = lockService.getLock(lockPath,
            LockService.LockMode.LOCK_EXCLUSIVE);
        if (!locked) {
          throw new IOException("Can't drop table:" + tableName
              + "(lock error:" + lockPath + ")");
        }
      } else {
        lockMetaData = new LockMetaData(lockPath);
        lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());

        boolean locked = lockService.createNode(lockMetaData, true);
        if (!locked) {
          throw new IOException("Can't drop table:" + tableName
              + "(lock error:" + lockPath + ")");
        }
      }

      // Live한 TabletServer에 drop 요청
      synchronized (dropingTabletServers) {
        List<String> dropTabletServerList = new ArrayList<String>();
        for (String hostName : runningTabletServers.keySet()) {
          TabletManagerProtocol tabletServer = null;
          try {
            tabletServer = connectTabletServer(hostName);
            if (tabletServer.dropTable(getTaskId(), tableName)) {
              dropTabletServerList.add(hostName);
            }
          } catch (IOException e) {
            LOG.fatal("TabletServer[" + hostName + "] can't drop table:"
                + tableName + "." + " shutdown tabletserver", e);
            try {
              tabletServer.shutdown();
            } catch (IOException err) {
              LOG.error("Can't shutdown tabletserver:" + hostName, err);
            }
          }
        }
        // 모든 TabletServer에서 처리가 완료될때까지 기다린다.
        // TODO 하나의 테이블이 drop 처리 중인 경우 다른 테이블은 Drop 되지 않는다.
        dropingTabletServers.put(tableName, dropTabletServerList);
        long startTime = System.currentTimeMillis();
        while (true) {
          dropingTabletServers.wait(10 * 1000);

          if (dropingTabletServers.get(tableName) == null) {
            break;
          }
          if (dropingTabletServers.get(tableName).size() == 0) {
            dropingTabletServers.remove(tableName);
            break;
          }

          // 5 min
          long gap = System.currentTimeMillis() - startTime;
          if (gap > 5 * 60 * 1000) {
            LOG.error(tableName + " Drop timeout after " + gap);
            addErrorTrace(tableName + " Drop timeout after " + gap);
            dropingTabletServers.remove(tableName);
            break;
          }
        }
      }

      String errorMessage = finalizeDrop(NeptuneMaster.this, conf, fs,
          lockService, tableName, false);
      if (errorMessage != null && errorMessage.length() > 0) {
        addErrorTrace(errorMessage);
      }
      LOG.debug("Drop table end:" + tableName);
    }
  }

  public static String finalizeDrop(NeptuneMaster neptuneMaster,
      NConfiguration conf, NeptuneFileSystem fs, LockService lockService,
      String tableName, boolean forced) throws IOException {
    String errorMessage = "";

    // Schema 정보 삭제
    boolean schemaDeleteResult = false;
    String schemaLocation = conf.get("table.schema.type", "filesystem");
    if(schemaLocation.equals("pleiades")) {
      schemaDeleteResult = lockService.deleteNode(Constants.PATH_SCHEMA + "/" + tableName, true);
    } else {
      GPath tableSchemaPath = TableSchema.getTableSchemaPath(conf, tableName);
      schemaDeleteResult = fs.delete(tableSchemaPath, true);
    }
    
    if (!schemaDeleteResult) {
      LOG.error("Can't delete schema:" + schemaDeleteResult);
      errorMessage += "Can't delete schema:" + schemaDeleteResult + "\n";
    }

    // Table 데이터를 Trash로 rename
    String tableDataPath = TableSchema.getTableDataPath(conf, tableName);
    if (fs.exists(new GPath(tableDataPath))) {
      String tableTrashPath = TableSchema.getTableDataTrashPath(conf, tableName);
      GPath dataTrashPath = new GPath(tableTrashPath);
      fs.mkdirs(dataTrashPath.getParent());
      boolean renameToResult = fs.renameTo(new GPath(tableDataPath),
          dataTrashPath);
      if (!renameToResult) {
        LOG.error("Can't rename table file:" + tableDataPath + " to "
            + tableTrashPath + " : " + renameToResult);
        // LOG.fatal("NeptuneMaster shutdown cause: Can't move table data file to trash while droping");
        errorMessage += "Can't rename table file:" + tableDataPath + " to "
            + tableTrashPath + " : " + renameToResult + "\n";
      }
    }
    // TODO ChangeLog도 삭제하지 않고 Trash로 이동

    // META 정보를 이용하여 처리되지 않은 Tablet의 META 삭제
    NTable ntable = NTable.openTable(conf, TABLE_NAME_META);
    TableScanner scanner = null;
    List<ScanCell> scanCells = new ArrayList<ScanCell>();
    try {
      scanner = ScannerFactory.openScanner(ntable, Tablet.generateMetaRowKey(
          tableName, Row.Key.MIN_KEY), Tablet.generateMetaRowKey(tableName,
          Row.Key.MAX_KEY), META_COLUMN_NAME_TABLETINFO, 20);

      ScanCell scanCell = null;
      while ((scanCell = scanner.next()) != null) {
        TabletInfo tabletInfo = new TabletInfo();
        tabletInfo.readFields(scanCell.getBytes());

        if (tableName.compareTo(tabletInfo.getTableName()) < 0) {
          break;
        }
        if (tableName.equals(tabletInfo.getTableName())) {
          scanCells.add(scanCell);
        }
      }
    } finally {
      if (scanner != null)
        scanner.close();
    }

    if (scanCells != null) {
      for (ScanCell eachCell : scanCells) {
        TabletInfo tabletInfo = new TabletInfo();
        tabletInfo.readFields(eachCell.getBytes());

        try {
          if (tabletInfo.getAssignedHostName() != null) {
            if (NeptuneMaster.connectTabletServer(conf,
                tabletInfo.getAssignedHostName()).dropTable(null, tableName)) {
              continue;
            }
          }
          // TabletServer에서 삭제 못시킨 내역만 META에서 삭제.
          Row.Key rowKey = Tablet.generateMetaRowKey(tabletInfo.getTableName(),
              tabletInfo.getEndRowKey());
          ntable.remove(rowKey, Constants.META_COLUMN_NAME_TABLETINFO,
              new Cell.Key(tabletInfo.getTabletName()));

          ChangeLogFileSystemIF changeLogFileSystem = ChangeLogFileSystem
              .getChangeLogFileSystem(conf, null, lockService);

          Tablet.deleteChangeLog(changeLogFileSystem,
              tabletInfo.getTableName(), tabletInfo.getTabletName());
        } catch (IOException err) {
          LOG.error("Can't delet META while table drop:" + tabletInfo + ":"
              + err.getMessage(), err);
          errorMessage += "Can't delet META while table drop:" + tabletInfo
              + ":" + err.getMessage() + "\n";
        }
      }
    }

    // drop lock 제거
    String tableDropLockPath = Constants.TABLE_DROP + "/" + tableName;

    if (lockService.exist(tableDropLockPath)) {
      long time = System.currentTimeMillis();
      while (true) {
        boolean deleteResult = lockService.deleteNode(tableDropLockPath, true);
        if (forced || deleteResult) {
          break;
        }

        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          return null;
        }

        if (System.currentTimeMillis() - time > 30 * 1000) {
          LOG.error("Can't delete table drop lock:" + tableDropLockPath);
          errorMessage += "Can't delete table drop lock:" + tableDropLockPath
              + "\n";
        }
      }
    }

    // Master에서 schema 삭제
    if (neptuneMaster != null) {
      synchronized (neptuneMaster.tabletInfoMonitor) {
        neptuneMaster.tables.remove(tableName);
      }
    }

    LOG.debug("End finalizeDrop:" + tableName);

    return errorMessage;
  }

  public void endTableDrop(String taskId, String hostName, String tableName)
      throws IOException {
    synchronized (dropingTabletServers) {
      LOG.debug("Receive endTableDrop from tabletserver:" + hostName + ","
          + tableName);
      List<String> servers = dropingTabletServers.get(tableName);
      if (server != null) {
        servers.remove(hostName);
      }
      dropingTabletServers.notifyAll();
    }
  }

  public void errorTableDrop(String taskId, String hostName, String tableName,
      String message) throws IOException {
    synchronized (dropingTabletServers) {
      LOG.debug("Receive errorTableDrop from tabletserver:" + hostName + ","
          + tableName);
      AsyncTask task = asyncTaskManager.getAsyncTask(taskId);
      if (task != null) {
        task.addErrorTrace(hostName + ":" + message);
      }

      List<String> servers = dropingTabletServers.get(tableName);
      if (server != null) {
        servers.remove(hostName);
      }
      dropingTabletServers.notifyAll();
    }
  }

  class MasterServerLockChecker implements Runnable {
    public void run() {
      while (true) {
        try {
          Thread.sleep(30 * 1000);
        } catch (InterruptedException e) {
          return;
        }
        try {
          if (!tryMasterLock()) {
            // Master의 lock을 가져오지 못한 경우 sleep 한 후 다시 요청해본다.
            try {
              Thread.sleep(30 * 1000);
              if (!acquireMasterLock()) {
                LOG.fatal("NeptuneMaster shutdown cause master lock expired");
                shutdown();
              }
            } catch (InterruptedException e) {
              return;
            }
          }
        } catch (IOException e) {
          LOG.error(e.getMessage(), e);
        }
      }
    }
  }

  public TabletInfo[] getTablets(String tableName) {
    synchronized (tabletInfoMonitor) {
      Map<String, TabletInfo> tablets = tabletInfos.get(tableName);
      if (tablets == null)
        return null;

      Collection<TabletInfo> tabletInfoList = tablets.values();
      return tabletInfoList.toArray(new TabletInfo[tabletInfoList.size()]);
    }
  }

  public TableSchema[] listTables() {
    Collection<TableSchema> tableList = tables.values();

    return tableList.toArray(new TableSchema[tableList.size()]);
  }

  public void endTabletAssignment(TabletInfo tabletInfo, boolean created)
      throws IOException {
    // endTabletAssignment 메소드가 RPC를 이용해서 call 되는 경우 handler에 의해 처리되지 않는다.
    // 따라서 내부적으로 handler에 의해 처리되는 메소드를 call한다.
    proxyObject.endTabletAssignmentForHandler(tabletInfo, created);
  }

  /**
   * Proxy를 이용해서 테스트 코드를 작성하기 위해 인터페이스로 빼서 처리, 실제는 endTabletAssignment 처리
   */
  public void endTabletAssignmentForHandler(TabletInfo tabletInfo,
      boolean created) throws IOException {
    // FIXME 처리 중 오류가 발생하면 어떻게 되나??
    // LOG.info("endTabletAssignment:" + tabletInfo + ", created:" + created);
    if (!assigningTablets.containsKey(tabletInfo.getTabletName())) {
      // assigningTablets에 없는 경우는 Tablet이 split된 경우이기 때문에 아무런 처리를 하지 않는다.
      LOG
          .info("receive endTabletAssignment for tablet which is not assigningTablet");
      return;
    }

    synchronized (tabletInfoMonitor) {
      Map<String, TabletInfo> tabletInfoMap = tabletInfos.get(tabletInfo
          .getTableName());
      if (tabletInfoMap == null) {
        tabletInfoMap = new HashMap<String, TabletInfo>(50);
        tabletInfos.put(tabletInfo.getTableName(), tabletInfoMap);
      }
      tabletInfoMap.put(tabletInfo.getTabletName(), tabletInfo);

      unassignedTablets.remove(tabletInfo.getTabletName());
      assigningTablets.remove(tabletInfo.getTabletName());
      // tsReportedTabletInfos.put(tabletInfo.getTableName(), tabletInfo);
    }

    if (TABLE_NAME_ROOT.equals(tabletInfo.getTableName())) {
      endRootTableAssignment = true;
    } else if (TABLE_NAME_META.equals(tabletInfo.getTableName())) {
      // FIXME META tablet이 여러개로 split된 경우
      // if (!clusterReady) {
      // setClusterReady(true);
      // // 클러스터가 처음 시작되어 ROOT -> META -> User Table 순서로 할당 되는 경우에만 수행
      // UserTabletAssignmentThread userTabletAssignmentThread = new
      // UserTabletAssignmentThread(tabletInfo);
      // userTabletAssignmentThread.start();
      // //addRunningThread(userTabletAssignmentThread);
      // }
    }
  }

  /**
   * Tablet할당 에러 발생시 다음과 같이 처리한다. 1. Tablet을 다른 서버로 이관하는 도중에 발생한 에러의 경우 기존 서버에
   * stop service가 요청되어 있는 상태인데 stop server를 해제시키고 계속 서비스 하도록한다.cancelStop 요청에도
   * 에러가 발생한 경우 unsigned tablet 목록에 추가한다. 2. 새로 생성된 Tablet인 경우 Tablet을 unsigned
   * tablet 목록에 추가한다.
   */
  public void errorTabletAssignment(String hostName, TabletInfo tabletInfo) {
    LOG.info("errorTabletAssignment:" + tabletInfo);

    if (TABLE_NAME_ROOT.equals(tabletInfo.getTableName())) {
      LOG
          .fatal("NeptuneMaster shutdown cause:NeptuneMaster stop cause by ROOT tablet assignment fail");
      shutdown();
      return;
    }

    if (!assigningTablets.containsKey(tabletInfo.getTabletName())) {
      // assigningTablets에 없는 경우는 Tablet이 split된 경우이기 때문에 아무런 처리를 하지 않는다.
      LOG
          .error("receive errorTabletAssignment for tablet which is not assigningTablet");
      return;
    }

    synchronized (tabletInfoMonitor) {
      TabletServerInfo tabletServerInfo = runningTabletServers.get(hostName);
      tabletServerInfo.subtractNumOfTablets();
      assigningTablets.remove(tabletInfo.getTabletName());
      if(!this.tables.containsKey(tabletInfo.getTableName())) {
        LOG.warn("Garbage tablet:" + tabletInfo);
        return;
      } else {
        unassignedTablets.put(tabletInfo.getTabletName(), tabletInfo);
        LOG.error("Added unassignedTablets:" + tabletInfo);
      }
    }
  }

  public long getProtocolVersion(String protocol, long clientVersion)
      throws IOException {
    if (protocol.equals(TableManagerProtocol.class.getName())) {
      return TableManagerProtocol.versionID;
    } else if (protocol.equals(TabletMasterProtocol.class.getName())) {
      return TabletMasterProtocol.versionID;
    } else {
      throw new IOException("Unknown protocol to tablet server: " + protocol);
    }
  }

  public boolean checkServer() {
    return true;
  }

  /**
   * 모든 TabletServe로부터 서비스하고 있는 Tablet 목록을 받는다.
   */
  private List<TabletInfo> getTabletsFromTabletServer() throws IOException {
    List<TabletInfo> tabletInfos = new ArrayList<TabletInfo>();

    Map<String, LockMetaData> servers = lockService.readDir(Constants.SERVER);

    // TabletServer로부터 현재 할당되어 있는 Tablet 목록을 받는다.
    List<String> tabletServers = new ArrayList<String>();
    if (servers == null) {
      return tabletInfos;
    }

    for (LockMetaData lockMetaData : servers.values()) {
      String lockId = lockMetaData.getLockId();
      if (!lockService.tryLock(lockId, LockService.LockMode.LOCK_EXCLUSIVE)) {
        String tabletServerHostName = NetworkUtil.fileNameToHostName(lockId
            .substring(Constants.SERVER.length() + 1));
        tabletServers.add(tabletServerHostName);
      }
    }

    for (String tabletServerHost : tabletServers) {
      try {
        TabletManagerProtocol tabletServer = connectTabletServer(tabletServerHost);
        TabletInfo[] tsTabletInfos = tabletServer.reportTablets();
        for (TabletInfo tabletInfo : tsTabletInfos) {
          // tsReportedTabletInfos.put(tabletInfo.getTabletName(), tabletInfo);
          tabletInfos.add(tabletInfo);
        }
      } catch (Exception e) {
        LOG.error("Can't receive tablet reports from:" + tabletServerHost, e);
      }
    }

    return tabletInfos;
  }

  public void reportTabletSplited(final TabletInfo targetTablet,
      final TabletInfo[] splitedTablets) {
    synchronized (tabletInfoMonitor) {
      try {
        removeTabletInAllVariable(targetTablet);

        Map<String, TabletInfo> tablets = tabletInfos.get(splitedTablets[0]
            .getTableName());
        if (tablets == null) {
          tablets = new HashMap<String, TabletInfo>(100);
          tabletInfos.put(splitedTablets[0].getTableName(), tablets);
        }
        tablets.put(splitedTablets[0].getTabletName(), splitedTablets[0]);
        tablets.put(splitedTablets[1].getTabletName(), splitedTablets[1]);
        // LOG.info("add splited tablet info:" +
        // splitedTablets[0].getTabletName() + "," +
        // splitedTablets[1].getTabletName());
      } catch (Exception e) {
        LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
        shutdown();
        return;
      }
    }

    Thread assignThread = new Thread(threadGroup, new Runnable() {
      public void run() {
        try {
          // 이전에 서비스하던 tablet server에서 stop이 완료된 후에 로딩 & 서비스하는 경우 false 전달
          assignTablet(splitedTablets[0]);
        } catch (Exception e) {
          // 예외가 발생할 경우 Master에서는 주기적으로 돌면서 로드 밸런싱 처리.
        } finally {
          // removeRunningThread(this);
        }
      }
    });
    assignThread.start();
  }

  // public void stopTabletService(TabletInfo stopTableInfo) throws IOException
  // {
  // connectTabletServer(stopTableInfo.getAssignedHostName()).stopTabletService(stopTableInfo);
  // }

  private void removeTabletInAllVariable(TabletInfo tabletInfo) {
    Map<String, TabletInfo> tablets = tabletInfos
        .get(tabletInfo.getTableName());
    if (tablets != null) {
      tablets.remove(tabletInfo.getTabletName());
    }
    unassignedTablets.remove(tabletInfo.getTabletName());
    assigningTablets.remove(tabletInfo.getTabletName());
  }

  public void shutdown() {
    if (!conf.getBoolean("testmode", false)) {
      System.exit(0);
    }

    try {
      LockServiceFactory.close(hostName);
    } catch (Exception e) {
    }

    if (!stopRequested) {
      stopRequested = true;
      server.stop();

      if (infoServer != null) {
        try {
          infoServer.stop();
        } catch (InterruptedException e1) {
        }
      }

      // ThreadGroup threadGroup = Thread.currentThread().getThreadGroup();
      Thread[] threads = new Thread[threadGroup.activeCount()];
      threadGroup.enumerate(threads, true);

      for (Thread thread : threads) {
        try {
          // LOG.debug("Master shutdown:child thread:" + thread.getId());
          thread.interrupt();
        } catch (Exception e) {
        }
      }
    }

    try {
      LockServiceFactory.close(hostName);
    } catch (Exception e) {
    }
    LOG.debug("shutdown masterserver:" + hostName);
  }

  public boolean isShutdowned() {
    return stopRequested;
  }

  /**
   * Master가 시작될 때 Root, Meta Tablet을 할당하고 Tablet 정보를 구성한다.
   * 
   * @author babokim
   */
  class RootTabletAssignmentThread implements Runnable {
    public void run() {
      // assignment root tablet
      // root 테이블을 생성한 다음 특정 tabletserver에 할당할이 완료될때 까지 반복한다.
      boolean stop = false;
      while (!stop) {
        try {
          Thread.sleep(500);
        } catch (InterruptedException e) {
          return;
        }
        try {
          stop = assignRootTablet();
        } catch (Exception e) {
          e.printStackTrace();
        }
      }

      // check finished root tablet assignment
      while (!endRootTableAssignment) {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          return;
        }
      }

      // report from tablet server. find unassigned tablet
      Map<String, TabletInfo> assignedTablets = new HashMap<String, TabletInfo>();
      try {
        List<TabletInfo> tsTabletInfos = getTabletsFromTabletServer();
        for (TabletInfo eachTabletInfo : tsTabletInfos) {
          assignedTablets.put(eachTabletInfo.getTabletName(), eachTabletInfo);
        }
      } catch (Exception e) {
        LOG.fatal(e);
        shutdown();
      }

      // scan root tablet finding all meta tablet list
      try {
        scanMetaTabletFromRoot();
      } catch (Exception e) {
        LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
        shutdown();
        return;
      }

      // assignment meta tablet
      try {
        assignMetaTablet(assignedTablets);
      } catch (IOException e) {
        LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
        shutdown();
        return;
      }

      // MetaTablet이 모두 할당되기를 기다린다.
      while (true) {
        synchronized (tabletInfoMonitor) {
          if (assigningTablets.size() == 0 && unassignedTablets.size() == 0) {
            break;
          }
        }
        try {
          Thread.sleep(1 * 1000);
        } catch (InterruptedException e) {
          return;
        }
      }

      clusterReady = true;

      // 할당되지 않은 Tablet을 찾는다.
      initUnassignedUserTablet(assignedTablets);

      // 할당되지 않은 Tablet 할당
      assignUserTablet();

      // Drop 미처리된 내역을 조회한다.
      completeDropedTable();

      // Split 도중 정리가 안된 Tablet에 대한 작업을 수행한다.
      finalizeSplitTemp();

      LOG.debug("End RootTabletAssignmentThread, TabletAssign thread start");
      Thread assignThread = new Thread(threadGroup, new Runnable() {
        public void run() {
          while (true) {
            try {
              Thread.sleep(10 * 1000);
            } catch (InterruptedException e) {
              return;
            }
            // LOG.debug("Assign Thread:unassignedTablets.size()=" +
            // unassignedTablets.size());
            assignUserTablet();
          }
        }
      });

      assignThread.start();
    }

    private void completeDropedTable() {
      // LOG.debug("Check drop while master init");
      try {
        String dropLockPath = Constants.TABLE_DROP;

        Map<String, LockMetaData> lockMetaDatas = lockService.readDir(
            dropLockPath);
        if (lockMetaDatas == null || lockMetaDatas.size() == 0) {
          return;
        }

        // 이미 존재하는 경우는 NeptuneMaster가 장애 발생으로 재 시작된 경우
        // 이 경우에는 drop 완료된 tablet에 대한 정보를 확인하여 전체 tablet 중에 drop 완료된 tablet 목록은
        // 지운다.
        for (LockMetaData lockMetaData : lockMetaDatas.values()) {
          String tableName = LockServiceFactory
              .getLockMetaDataPathName(lockMetaData);
          // LOG.debug("Find drop lock:" + tableName);
          // if(isCompleteDrop(tableName)) {
          // //Drop 처리 완료된 경우
          // Thread dropThread = new Thread(threadGroup, new
          // DropThread(tableName));
          // dropThread.start();
          // } else {
          // Drop 처리 진행 중인 경우 다시 TableDrop 이벤트 발생
          //dropTable(tableName);
          asyncTaskManager.runAsyncTask(new TableDropTask(tableName));
          // }
        }

      } catch (Exception e) {
        LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
        shutdown();
      }
    }

    private void assignUserTablet() {
      Map<String, TabletInfo> assignTargetTablets = new HashMap<String, TabletInfo>(
          10);

      synchronized (tabletInfoMonitor) {
        assignTargetTablets.putAll(unassignedTablets);
      }

      for (TabletInfo eachTabletInfo : assignTargetTablets.values()) {
        try {
          proxyObject.assignTablet(eachTabletInfo);
        } catch (IOException e) {
          // e.printStackTrace();
        }
      }
    }

    private void initUnassignedUserTablet(
        Map<String, TabletInfo> assignedTablets) {
      List<TabletInfo> tempMetaTablets = new ArrayList<TabletInfo>();

      synchronized (tabletInfoMonitor) {
        Map<String, TabletInfo> metaTablets = tabletInfos.get(TABLE_NAME_META);
        tempMetaTablets.addAll(metaTablets.values());
      }

      for (TabletInfo metaTabletInfo : tempMetaTablets) {
        TableScanner scanner = null;
        try {
          try {
            scanner = ScannerFactory.openScanner(conf, metaTabletInfo, META_COLUMN_NAME_TABLETINFO);
          } catch (Exception e) {
            return;
          }

          ScanCell scanCell = null;
          while ((scanCell = scanner.next()) != null) {
            TabletInfo userTabletInfo = new TabletInfo();
            userTabletInfo.readFields(scanCell.getBytes());

            // LOG.debug("SCAN META for finding unassign tablet: " +
            // userTabletInfo);
            String userTableName = userTabletInfo.getTableName();

            Map<String, TabletInfo> tablets = tabletInfos.get(userTableName);
            if (tablets == null) {
              tablets = new HashMap<String, TabletInfo>();
              tabletInfos.put(userTableName, tablets);
            }
            tablets.put(userTabletInfo.getTabletName(), userTabletInfo);

            if (!assignedTablets.containsKey(userTabletInfo.getTabletName())) {
              synchronized (tabletInfoMonitor) {
                // LOG.debug("find unassigned tablet:" + userTabletInfo);
                unassignedTablets.put(userTabletInfo.getTabletName(),
                    userTabletInfo);
                // assignment는 TabletAssignmentThread에서 수행, 여기서는 unassignment로
                // 설정만 변경
                // proxyObject.assignTablet(userTabletInfo, true);
              }
            }
          }
        } catch (Exception e) {
          LOG.error("initUnassignedUserTablet: " + metaTabletInfo, e);
        } finally {
          if (scanner != null) {
            try {
              scanner.close();
            } catch (IOException e) {
              e.printStackTrace();
            }
          }
        }
      }
    }

    /**
     * ROOT tablet을 scan하여 META tablet의 목록을 가져온다. tabletInfos 멤버변수에 추가
     * 
     * @throws IOException
     */
    private void scanMetaTabletFromRoot() throws IOException {
      // LOG.debug("Start scanMetaTabletFromRoot: ");
      Map<String, TabletInfo> metaTabletInfos = tabletInfos
          .get(TABLE_NAME_META);
      if (metaTabletInfos == null) {
        metaTabletInfos = new HashMap<String, TabletInfo>(100);
        tabletInfos.put(TABLE_NAME_META, metaTabletInfos);
      }

      TabletInfo rootTabletInfo = tabletInfos.get(TABLE_NAME_ROOT).values()
          .iterator().next();
      // LOG.debug("scanMetaTabletFromRoot: " + rootTabletInfo);

      // Root Tablet이 최초 생성된 후에는 Meta Tablet이 없다.
      TableScanner scanner = ScannerFactory.openScanner(conf,
          rootTabletInfo, META_COLUMN_NAME_TABLETINFO);
      try {
        ScanCell scanCell = null;
        while ((scanCell = scanner.next()) != null) {
          TabletInfo metaTabletInfo = new TabletInfo();
          metaTabletInfo.readFields(scanCell.getBytes());

          metaTabletInfos.put(metaTabletInfo.getTabletName(), metaTabletInfo);
        }
      } finally {
        scanner.close();
      }
    }
  }// end of RootTabletAssignmentThread class

  /**
   * 일정시간동안 TabletServer로부터 Heartbeat 메세지가 오지 않으면 TabletServer가 disable된 것으로
   * 판단하여 해당 TabletServer에서 서비스 중인 Tablet을 다른 TabletServer에 할당한다. 이 Thread에서는 실제
   * 할당하는 것까지 수행하지는 않고 unassignedTablet에 추가만 한다.
   * 
   * @author babokim
   */
  class ReassignmentWhenTabletServerFailThread implements Runnable {
    // FIXME TabletServer fail 처리 완료되지 않은 상태에서 똑같은 서버가 다시 살아나거나
    // 살아난 후 다시 fail 되는 경우에 대한 처리
    private TabletServerInfo tabletServerInfo;

    public ReassignmentWhenTabletServerFailThread(
        TabletServerInfo tabletServerInfo) {
      this.tabletServerInfo = tabletServerInfo;
    }

    public void run() {
      String failedHostName = tabletServerInfo.getHostName();

      LOG.debug("Failed TabletServer:" + failedHostName + ", time="
          + System.currentTimeMillis() + ", Master table count="
          + tabletInfos.size());

      // Split 처리중 META는 수정 전/후에 TabletServer가 fail 된 경우 NeptuneMaster도 자동으로
      // 죽는다.
      // 이유는 Tablet 정보가 불일치 하기 때문
      String splitLockPath = Constants.TABLETSERVER_SPLIT + "/"
          + NetworkUtil.hostNameToFileName(failedHostName);
      int retry = 0;
      while(true) {
        try {
          Map<String, LockMetaData> splitedTabletMetaDatas = lockService.readDir(splitLockPath);
          if (splitedTabletMetaDatas != null && splitedTabletMetaDatas.size() > 0) {
            LOG
                .fatal("NeptuneMaster shutdown cause: tablet server failed while spliting");
            shutdown();
            return;
          } else {
            break;
          }
        } catch (IOException e1) {
          retry++;
          if(retry > 10) {
            LOG.fatal("NeptuneMaster shutdown cause: Eror calling readDir(" + splitLockPath + ") while ReassignmentWhenTabletServerFailThread");
            shutdown();
            return;
          }
          LOG.warn("Eror calling readDir(" + splitLockPath + ")" + e1.getMessage() + ", but retry:" + retry);
          try {
            Thread.sleep(2000);
          } catch (InterruptedException e) {
            return;
          }
        }
      }

      // FIXME tabletInfos가 많은 경우 synchronized 블럭이 너무 오래 수행된다.
      TabletInfo rootTablet = null;
      Set<TabletInfo> metaTabletInfos = new HashSet<TabletInfo>();
      Set<TabletInfo> userTabletInfos = new HashSet<TabletInfo>();

      synchronized (tabletInfoMonitor) {
        // Master server에서 관리 중인 tablet 목록에서 fail된 서버의 tablet server에서 관리하는
        // tablet을 찾는다.
        for (Map<String, TabletInfo> tablets : tabletInfos.values()) {
          for (TabletInfo tabletInfo : tablets.values()) {
            if (failedHostName.equals(tabletInfo.getAssignedHostName())) {
              if (TABLE_NAME_ROOT.equals(tabletInfo.getTableName())) {
                rootTablet = tabletInfo;
                // LOG.debug("root Tablet on the failed TabletServer(tabletInfos):"
                // + tabletInfo);
              } else if (TABLE_NAME_META.equals(tabletInfo.getTableName())) {
                metaTabletInfos.add(tabletInfo);
                // LOG.debug("meta Tablet on the failed TabletServer(tabletInfos):"
                // + tabletInfo);
              } else {
                userTabletInfos.add(tabletInfo);
                // LOG.debug("user Tablet on the failed TabletServer(tabletInfos):"
                // + tabletInfo);
              }
              LOG.debug("Tablet on the failed TabletServer(tabletInfos):"
                  + tabletInfo);
              assigningTablets.remove(tabletInfo.getTabletName());
            }
          }
        }

        // 현재 할당 중인 tablet 목록에서 fail된 서버의 tablet server에서 관리하는 tablet을 찾는다.
        for (TabletInfo tabletInfo : assigningTablets.values()) {
          if (failedHostName.equals(tabletInfo.getAssignedHostName())) {
            if (TABLE_NAME_ROOT.equals(tabletInfo.getTableName())) {
              rootTablet = tabletInfo;
              // LOG.debug("root Tablet on the failed TabletServer(assigningTablets):"
              // + tabletInfo);
            } else if (TABLE_NAME_META.equals(tabletInfo.getTableName())) {
              metaTabletInfos.add(tabletInfo);
              // LOG.debug("meta Tablet on the failed TabletServer(assigningTablets):"
              // + tabletInfo);
            } else {
              userTabletInfos.add(tabletInfo);
              // LOG.debug("user Tablet on the failed TabletServer(assigningTablets):"
              // + tabletInfo);
            }
            // assigningTablets.remove(tabletInfo.getTabletName());
            LOG.debug("Tablet on the failed TabletServer(assigningTablets):"
                + tabletInfo);
          }
        }

        // 현재 drop 처리 중인 테이블의 Tablet의 할당 목록에서 삭제한다.
        List<TabletInfo> removeTargets = new ArrayList<TabletInfo>();
        for (TabletInfo tabletInfo : assigningTablets.values()) {
          if (dropingTabletServers.containsKey(tabletInfo.getTableName())) {
            removeTargets.add(tabletInfo);
          }
        }
        for (TabletInfo tabletInfo : removeTargets) {
          assigningTablets.remove(tabletInfo.getTabletName());
        }

        removeTargets = new ArrayList<TabletInfo>();
        for (TabletInfo tabletInfo : userTabletInfos) {
          if (dropingTabletServers.containsKey(tabletInfo.getTableName())) {
            removeTargets.add(tabletInfo);
          }
        }
        for (TabletInfo tabletInfo : removeTargets) {
          userTabletInfos.remove(tabletInfo.getTabletName());
        }
      }

      // ROOT Tablet 재할당
      if (rootTablet != null) {
        try {
          assignTablet(rootTablet);
        } catch (Exception e) {
          LOG.error("ReassignmentWhenTabletServerFailThread: assign ROOT");
        }
      }

      try {
        // //////////////////////////////////////////////////////
        // META Tablet 재할당
        for (TabletInfo tabletInfo : metaTabletInfos) {
          try {
            assignTablet(tabletInfo);
          } catch (Exception e) {
            LOG.error("ReassignmentWhenTabletServerFailThread: assign META");
          }
        }

        // //////////////////////////////////////////////////////
        // User Tablet 재할당
        for (TabletInfo tabletInfo : userTabletInfos) {
          try {
            // synchronized(tabletInfoMonitor) {
            // unassignedTablets.put(tabletInfo.getTabletName(), tabletInfo);
            // }
            assignTablet(tabletInfo);
            // LOG.debug("ReassignmentWhenTabletServerFailThread: assign request: "
            // + tabletInfo);
          } catch (Exception e) {
            LOG
                .error("ReassignmentWhenTabletServerFailThread: assign userTabletInfos: "
                    + tabletInfo);
          }
        }
      } catch (Exception e) {
        LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
        shutdown();
      }
    }
  }

  /**
   * TabletServer에서 Split 도중 마지막 단계인 Master로 report하기 전/후로 장애가 발생한 경우에 대한 처리
   * Split은 정상적으로 되었기 때문에 이전에 있던 tablet의 파일과 정보를 삭제하는 기능을 수행한다.
   * 
   * @throws IOException
   */
  private void finalizeSplitTemp() {
    Map<String, LockMetaData> splitHostInfos = null;
    try {
      splitHostInfos = lockService.readDir(Constants.TABLETSERVER_SPLIT);
      if (splitHostInfos == null || splitHostInfos.size() == 0) {
        return;
      }
    } catch (IOException e) {
      LOG.warn("finalizeSplitTemp lockService.readDir error", e);
    }

    try {
      for (LockMetaData eachLockMetaData : splitHostInfos.values()) {
        String splitHostLockId = eachLockMetaData.getLockId();
        String hostName = NetworkUtil.fileNameToHostName(LockServiceFactory
            .getLockMetaDataPathName(eachLockMetaData));
        LOG.info("finalizeSplitTemp:host=" + hostName);

        Map<String, LockMetaData> tabletSplitInfos = lockService
            .readDir(splitHostLockId);
        if (tabletSplitInfos == null || tabletSplitInfos.size() == 0) {
          lockService.deleteNode(splitHostLockId, true);
          continue;
        }

        for (LockMetaData splitMetaData : tabletSplitInfos.values()) {
          String targetTabletName = LockServiceFactory
              .getLockMetaDataPathName(splitMetaData);
          LOG.info("finalizeSplitTemp:tablet=" + targetTabletName);
          Object[] infos = parseMetaInfos(splitMetaData.getContentKey());

          String tableName = (String) infos[0];
          // 접속을 해서 해당 Tablet이 존재하는지 확인해본다.
          try {
            TabletInfo tabletInfo = connectTabletServer(hostName)
                .getTabletInfo(targetTabletName);
            if (tabletInfo != null) {
              LOG.fatal("Tablet Split aborted abnormally." + targetTabletName
                  + "," + infos[1] + "," + infos[2]);
              shutdown();
            }
          } catch (Exception e) {
            // 장애가 발생한 경우나 Tablet 정보가 존재하지 않는 경우
          }

          Map<String, TabletInfo> managedTabletInfos = tabletInfos
              .get(tableName);
          if (managedTabletInfos != null
              && managedTabletInfos.containsKey(targetTabletName)) {
            LOG.fatal("Tablet Split aborted abnormally." + targetTabletName
                + "," + infos[1] + "," + infos[2]);
            shutdown();
          } else {
            // 해당 Tablet의 모든 파일 삭제
            GPath dataPath = Tablet.getTabletPath(conf, tableName,
                targetTabletName);
            LOG.info("delete tablet files while finalize split:"
                + targetTabletName + ":" + dataPath);
            long startTime = System.currentTimeMillis();
            while (true) {
              boolean result = fs.delete(dataPath, true);
              if (result) {
                break;
              }
              try {
                Thread.sleep(100);
              } catch (InterruptedException e) {
                return;
              }
              if (System.currentTimeMillis() - startTime > 30 * 1000) {
                LOG.warn("Can't delete " + dataPath + " while droping "
                    + targetTabletName);
                break;
              }
            }
          }
        } // for each tablet
        lockService.deleteNode(splitHostLockId, true);
      } // for each tabletserver
    } catch (Exception e) {
      LOG.fatal("NeptuneMaster shutdown cause:" + e.getMessage(), e);
      shutdown();
    }
  }

  /**
   * 0: tableName, 1,2: tabletInfo
   * 
   * @param metaInfo
   * @return
   */
  private Object[] parseMetaInfos(final String metaInfo) {
    Object[] result = new Object[3];

    String tempMetaInfo = metaInfo;
    int length = Integer.parseInt(tempMetaInfo.substring(0, 5));
    result[0] = metaInfo.substring(5, length + 5); // table Name

    for (int i = 1; i < 3; i++) {
      tempMetaInfo = tempMetaInfo.substring(5 + length);
      length = Integer.parseInt(tempMetaInfo.substring(0, 5));
      String tabletName = tempMetaInfo.substring(5, length + 5);

      tempMetaInfo = tempMetaInfo.substring(5 + length);
      length = Integer.parseInt(tempMetaInfo.substring(0, 5));
      String endRowKey = tempMetaInfo.substring(5, length + 5);

      result[i] = new TabletInfo((String) result[0], tabletName, null,
          new Row.Key(endRowKey));
    }

    return result;

    // df.format(targetTablet.getTabletInfo().getTableName().getBytes().length)
    // +
    // targetTablet.getTabletInfo().getTableName() +
    // df.format(splitedTablets[0].getTabletInfo().getTabletName().getBytes().length)
    // +
    // splitedTablets[0].getTabletInfo().getTabletName() +
    // df.format(splitedTablets[0].getTabletInfo().getEndRowKey().getBytes().length)
    // +
    // splitedTablets[0].getTabletInfo().getEndRowKey() +
    // df.format(splitedTablets[1].getTabletInfo().getTabletName().getBytes().length)
    // +
    // splitedTablets[1].getTabletInfo().getTabletName() +
    // df.format(splitedTablets[1].getTabletInfo().getEndRowKey().getBytes().length)
    // +
    // splitedTablets[1].getTabletInfo().getEndRowKey();
  }

  private void addRunningTabletServer(String tabletServerHostName)
      throws IOException {
    LOG.info("Notice tablet server created: " + tabletServerHostName);

    TabletServerInfo tabletServerInfo = new TabletServerInfo(
        tabletServerHostName);
    synchronized (runningTabletServers) {
      if (runningTabletServers.containsKey(tabletServerHostName)) {
        return;
      }
      // call release all tablets. if tablet server reconnected.
      // connectTabletServer(tabletServerInfo).stopAllTablets();
      runningTabletServers.put(tabletServerHostName, tabletServerInfo);
    }
  }


  public void reportTabletServerStarted(String hostName) throws IOException {
    addRunningTabletServer(hostName);
  }
  
  private void addLockEventHandler() throws IOException {
    // TabletServer가 추가되었을 경우
    LockEventMonitor tabletServerAddEventMonitor = new FilteredLockEventMonitor() {
      private LockEventFilter filter;

      public synchronized LockEventFilter getFilter() {
        if (filter == null) {
          filter = new LockEventFilter();
          filter.addConstraint(LockEventFilter.Operator.PF, Constants.SERVER);

          List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
          lockEvents.add(LockService.Events.NODE_ADDED);
          filter.setLockEventList(lockEvents);
        }
        return filter;
      }

      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        try {
          if (!tryMasterLock()) {
            return;
          }
          String lockId = eventData.getLockId();
          String tabletServerHostName = NetworkUtil.fileNameToHostName(lockId
              .substring(Constants.SERVER.length() + 1));
          addRunningTabletServer(tabletServerHostName);
        } catch (Exception e) {
          LOG.error(e);
        }
      }
    };

    if(lockService.getEventHandler() != null) {
      lockService.getEventHandler().addLockEventMonitor(
          tabletServerAddEventMonitor);
    }

    // TabletServer에 장애가 발생했을 경우
    LockEventMonitor tabletServerRemovedEventMonitor = new LockEventMonitor() {
      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, Constants.SERVER);

        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.LOCK_EXPIRED);
        filter.setLockEventList(lockEvents);

        return filter;
      }

      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        String lockId = eventData.getLockId();
        String tabletServerHostName = NetworkUtil.fileNameToHostName(lockId
            .substring(Constants.SERVER.length() + 1));
        LOG.info("Received tablet server failed event: " + lockId + ","
            + tabletServerHostName);

        try {
          if (!tryMasterLock()) {
            return;
          }
        } catch (Exception e) {
          LOG.error(e);
          return;
        }

        TabletServerInfo tabletServerInfo = new TabletServerInfo(
            tabletServerHostName);
        String tabletServerLockPath = Constants.SERVER + "/"
            + NetworkUtil.hostNameToFileName(tabletServerHostName);

        //TabletServer에 대한 lock을 시도해본다. lock이 되면 장애가 발생한 상황
        boolean locked = false;
        int tryCount = 0;
        while(true) {
          try {
            locked = lockService.getLock(tabletServerLockPath,
              LockService.LockMode.LOCK_EXCLUSIVE);
            break;
          } catch (IOException e) {
            tryCount++;
            if (tryCount >= 10) {
              LOG.fatal("shutdown cause getLock error while tabletserver fail processing:" + tabletServerHostName, e);
              shutdown();
            }   
            LOG.warn("getLock error while tabletserver fail processing:" + e.getMessage() + ", but retry:" + tryCount);
            try {
              Thread.sleep(2000);
            } catch (InterruptedException e1) {
              return;
            }
          }
        }
        
        if (locked) {
          tryCount = 0;
          while (tryCount < 5) {
            try {
              if (lockService.deleteNode(tabletServerLockPath, true)) {
                break;
              }
            } catch (IOException e1) {
              LOG.warn("deleteNode error while tabletserver fail processing:" + e1.getMessage() + ", but retry:" + tryCount);
            }
            //삭제하지 못한 경우 다시 시도한다.
            try {
              Thread.sleep(1000);
            } catch (InterruptedException e) {
              return;
            }
            tryCount++;
          }
          if (tryCount >= 5) {
            LOG.error("Received tablet server[" + tabletServerHostName
                + "] fail message. But can't delete server lock.");
          }
        } else {
          // TabletServer가 다시 클러스터에 붙은 경우이기 때문에 Fail 처리하지 않는다.
          return;
        }

        synchronized (runningTabletServers) {
          if (!runningTabletServers.containsKey(tabletServerHostName)) {
            return;
          }
          LOG.info("Notice tablet server failed: " + lockId + ","
              + tabletServerHostName);
          runningTabletServers.remove(tabletServerHostName);
        }

        Thread thread = new Thread(threadGroup,
            new ReassignmentWhenTabletServerFailThread(tabletServerInfo));
        thread.start();

        // Drop 처리 중인 TabletServer의 목록에서 삭제
        synchronized (dropingTabletServers) {
          for (List<String> entry : dropingTabletServers.values()) {
            entry.remove(tabletServerHostName);
          }
        }
      }
    };
    if(lockService.getEventHandler() != null) {
      lockService.getEventHandler().addLockEventMonitor(
          tabletServerRemovedEventMonitor);
    }

    // scan live tablet server(event 핸들러 등록 전에 add된 TabletServer를 가져온다.
    Map<String, LockMetaData> servers = lockService.readDir(Constants.SERVER);

    if (servers == null) {
      return;
    }

    for (LockMetaData lockMetaData : servers.values()) {
      String lockId = lockMetaData.getLockId();
      if (!lockService.tryLock(lockId, LockService.LockMode.LOCK_EXCLUSIVE)) {
        String tabletServerHostName = NetworkUtil.fileNameToHostName(lockId
            .substring(Constants.SERVER.length() + 1));
        addRunningTabletServer(tabletServerHostName);
      }
    }

    // master lock이 expired 되었을 경우
    LockEventMonitor sessionExipreEventMonitor = new LockEventMonitor() {

      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF,
            Constants.MASTER_SERVER);

        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.SESSION_EXPIRED);
        filter.setLockEventList(lockEvents);

        return filter;
      }

      // Session이 expire되는 경우 네트워크 문제 등으로 인해 Pleiades에 lock을 유지 못하는 경우이다.
      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        try {
          Thread.sleep(30 * 1000);
        } catch (InterruptedException e) {
          return;
        }
        try {
          if (!acquireMasterLock()) {
            LOG.fatal("NeptuneMaster shutdown cause master lock expired");
            shutdown();
          }
        } catch (IOException e) {
          LOG.fatal("NeptuneMaster shutdown cause master lock expired2", e);
          shutdown();
        }
      }
    };
    if(lockService.getEventHandler() != null) {
      lockService.getEventHandler()
          .addLockEventMonitor(sessionExipreEventMonitor);
    }
  }
  
  public String getHostName() {
    return this.hostName;
  }

  public String getTestHandlerKey() {
    return hostName;
  }

  public void test() {
    LOG.debug("called test:" + hostName);
  }

  public NConfiguration getConf() {
    return conf;
  }

  public TabletServerInfo[] getTabletServerInfos() {
    synchronized (runningTabletServers) {
      if (runningTabletServers.size() == 0) {
        return null;
      }
      return runningTabletServers.values().toArray(new TabletServerInfo[] {});
    }
  }

  public void addUser(String userId) throws IOException {
    // 여기에 전달되는 passwd는 평문
    if (!AclManager.isSuperGroup(lockService)) {
      throw new IOException("Can't access user management API");
    }
    addUser(lockService, userId);
  }

  private static void addUser(LockService lockService, String userId)
      throws IOException {
    LockMetaData userLockMetaData = new LockMetaData(USERS + "/" + userId);
    userLockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
    userLockMetaData.setContentKey(userId);
    lockService.createNode(userLockMetaData, true);

    lockService.releaseLock(USERS + "/" + userId,
        LockService.LockMode.LOCK_EXCLUSIVE);
  }

  private static void addSuperGroupUser(LockService lockService, String userId)
      throws IOException {
    LockMetaData userLockMetaData = new LockMetaData(SUPERGROUP);
    userLockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
    userLockMetaData.setContentKey(userId);
    lockService.createNode(userLockMetaData, true);

    lockService.releaseLock(SUPERGROUP, LockService.LockMode.LOCK_EXCLUSIVE);
  }

  private boolean existSuperGroup() throws IOException {
    return lockService.exist(SUPERGROUP);
  }

  public void removeUser(String userId) throws IOException {
    if (!AclManager.isSuperGroup(lockService)) {
      throw new IOException("Can't access user management API");
    }

    if (!lockService.getLock(USERS + "/" + userId,
        LockService.LockMode.LOCK_EXCLUSIVE)) {
      throw new IOException("Can't get lock:" + USERS + "/" + userId);
    }

    lockService.deleteNode(USERS + "/" + userId, true);
  }

  public void addTablePermission(String tableName, String userId,
      String readWrite) throws IOException {
    AclManager.checkOwner(conf, lockService, tables, tableName);

    synchronized (tables) {
      if (!tables.containsKey(tableName)) {
        throw new IOException("No table:" + tableName);
      }

      TableSchema tableSchema = tables.get(tableName);
      tableSchema.addPermission(userId, readWrite);
      saveTablePermission(tableSchema);
    }
  }

  public void removeTablePermission(String tableName, String userId)
      throws IOException {
    AclManager.checkOwner(conf, lockService, tables, tableName);

    synchronized (tables) {
      if (!tables.containsKey(tableName)) {
        throw new IOException("No table:" + tableName);
      }

      TableSchema tableSchema = tables.get(tableName);
      tableSchema.removePermission(userId);
      saveTablePermission(tableSchema);
    }
  }

  private void saveTablePermission(TableSchema tableSchema) throws IOException {
    String path = PATH_SCHEMA + "/" + tableSchema.getTableName() + SCHEMA_DETAIL;
    if (!lockService.getLock(path, LockService.LockMode.LOCK_EXCLUSIVE)) {
      throw new IOException("Can't get lock:" + path);
    }

    lockService.setContents(path, tableSchema.getOwnerInfo());
    lockService.releaseLock(path, LockService.LockMode.LOCK_EXCLUSIVE);
  }

  private boolean existUser(String userId) throws IOException {
    String path = USERS + "/" + userId;
    return lockService.exist(path);
  }
}
